
#! /usr/bin/pytohn
# -*- coding: utf-8 -*- 


import os
import sys
import time
import threading
import pickle
from kafka import*

from collections import deque
import app_global
from dataproc_plc import *
from dataproc_shp import *

log=app_global.gLog

class CMsgConsumer(threading.Thread):

    def __init__(self,db,model):
        threading.Thread.__init__(self)
        self.db = db
        self.model=model
        self.kfk_c = KafkaConsumer(bootstrap_servers=['0.0.0.0:9092'],auto_offset_reset='latest')
        self.kfk_c.subscribe(['data_acq'])

        self.kfk_p = KafkaProducer(bootstrap_servers=['0.0.0.0:9092'])

   
    def run(self):
        self.q = deque(maxlen=10)
        while app_global.SYS_RUN:
            if self.kfk_c:
                self.kfk_c.poll(timeout_ms=5)
                for msg in self.kfk_c:
                    key = msg.key.decode(encoding="utf-8")               #因为接收到的数据时bytes类型，因此需要解码
                    value = pickle.loads(msg.value)

                    if key=='plc_data':
                        proc = CDataprocPlc(self.db,self.model,self.kfk_p)
                        proc.data_proc(value,self.q)
                    elif key=='shp_data':
                        proc=CDataprocShp(self.db)
                        proc.data_proc(value)
                    else:
                        print("key = {}".format(key))


        time.sleep(1)


if __name__=="__main__":
    #kafka consume
    #

    pass
